package com.tianqu.tms.core.webservice.weichat.api;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import com.tianqu.tms.core.base.Logger;
import com.tianqu.tms.core.webservice.weichat.api.rpc.AbstractBo;



/**
 * @author HuangWei
 *
 */
public class PullingExecutor {
	private Logger LOGGER = Logger.getLogger(PullingExecutor.class);
	private static volatile PullingExecutor instance;
	
	protected BlockingQueue<AbstractBo> queue;
	protected ThreadFactory threadFactory;
	protected long timeout = TimeUnit.SECONDS.toMillis(10);
	protected volatile boolean terminated;
	private int threadSize = 5;
	
	
    private PullingExecutor(){}
    
    private void doStart(){
    	this.terminated = false;
    	if(this.queue == null) this.queue = new LinkedBlockingQueue<AbstractBo>();
    	for(int i=0;i<threadSize;i++) new Thread(new Puller()).start();
    }
    
    public void doStop(){
    	this.terminated = true;
    }
    
    public boolean submit(final AbstractBo work) {
		return queue.offer(work);
	}
    
    protected class Puller implements Runnable {

		@Override
		public void run() {
			try {
				while(!terminated) {
					AbstractBo work = queue.poll(50,TimeUnit.NANOSECONDS);
					if(work == null) continue; WeChatMqApi.getInstance().call(work);
				}
			} catch (InterruptedException e) {
				LOGGER.error("Puller error",e);
			}
			
			
			LOGGER.error("PullingExecutor is stop!");
		}
    }
    
    public static PullingExecutor getInstance() {
    	if (instance == null) {    
            synchronized (PullingExecutor.class) {    
               if (instance == null) {    
            	   instance = new PullingExecutor();   
            	   instance.doStart();
               }    
            }    
         }   
        return instance; 
	}
     
}
